package com.TanNgee.netty.kyro.client;

/**
 * @author 陈毅
 * @create 2022/3/11 0:06
 */

//https://www.cnblogs.com/lbhym/p/12753314.html

import com.TanNgee.netty.kyro.codec.NettyKryoDecoder;
import com.TanNgee.netty.kyro.codec.NettyKryoEncoder;
import com.TanNgee.netty.kyro.dto.RpcRequest;
import com.TanNgee.netty.kyro.dto.RpcResponse;
import com.TanNgee.netty.kyro.serialize.KryoSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 客户端
 * 通过这个方法，可以将RpcRequest 对象发送到服务端
 */
public class NettyClient {
    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
    private final String host;
    private final int port;
    private static final Bootstrap b;


 /*
    Server 创建连接过程
        1. 指定线程模型
        2.指定 IO 类型为 NIO
        3.IO 处理逻辑
        4.建立连接
   */

    public NettyClient(String host, int port) {
        this.port = port;
        this.host = host;
    }

    //初始化相关资源，EventLoopGroup，Bootstarp
    static {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); //（事件循环）
        b = new Bootstrap(); //（启动引导类）

        //kryo对象
        KryoSerializer kryoSerializer = new KryoSerializer();
        // 1. 指定线程模型
        b.group(eventLoopGroup)
                // 2. channel是配置客户端的IO模型，配置的是NIO模型
                .channel(NioSocketChannel.class)

                // 连接的超时时间，超过这个时间还是建立不上的话则代表连接失败
                //  如果 15 秒之内没有发送数据给服务端的话，就发送一次心跳请求
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)

                .handler(new LoggingHandler(LogLevel.INFO))
                // 3.IO 处理逻辑
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        /*
                         自定义序列化编解码器
                         */
                        // RpcResponse -> ByteBuf
                        ch.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcResponse.class));
                        // ByteBuf -> RpcRequest
                        ch.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcRequest.class));

                        ch.pipeline().addLast(new NettyClientHandler());
                    }
                });


    }

    /**
     * 发送消息到服务端
     *
     * @param rpcRequest
     * @return 服务端返回的数据
     */
    public RpcResponse sendMessage(RpcRequest rpcRequest) {
        try {
            ChannelFuture f = b.connect(host, port).sync(); // 4.建立连接

            logger.info("client connect  {}", host + ":" + port);
            Channel futureChannel = f.channel();
            logger.info("send message");

            if (futureChannel != null) {
                futureChannel.writeAndFlush(rpcRequest).addListener(future -> {
                    if (future.isSuccess()) { //连接成功
                        logger.info("client send message: [{}]", rpcRequest.toString());
                    } else { // 连接失败
                        logger.error("Send failed:", future.cause());
                    }
                });
                futureChannel.closeFuture().sync();
                AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
                return futureChannel.attr(key).get();
            }
        } catch (InterruptedException e) {
            logger.error("occur exception when connect server:", e);
        }
        return null;

    }

    public static void main(String[] args) {
        // 消息体
        RpcRequest rpcRequest = RpcRequest.builder()
                .interfaceName("interface")
                .methodName("hello").build();

        // 发起连接
        NettyClient nettyClient = new NettyClient("127.0.0.1", 8889);

        // 发送消息
        for (int i = 0; i < 3; i++) {
            nettyClient.sendMessage(rpcRequest);
        }
        // 接收返回消息
        RpcResponse rpcResponse = nettyClient.sendMessage(rpcRequest);
        System.out.println(rpcResponse.toString());
    }
}
